package com.niit.covid.sum;


import com.niit.covid.bean.CovidCountBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author:Ys
 * @date: 2022年08月11日 14:03
 * @desc:
 *      需求：统计每个州累计确诊病例。
 *      分析：
 *          自定义一个对象CovidCountBean，用于封装每个县的确诊病例数和死亡病例数。注意需要实现hadoop的序列化机制。
 *          以州state作为map阶段输出的key,以CovidCountBean作为value，这样经过MapReduce的默认排序分组规则，属于同一个州的数据就会变成一组进行reduce处理，进行累加即可得出每个州累计确诊病例。
 */
public class CovidSumMapper extends Mapper<LongWritable, Text, Text, CovidCountBean> {

    /**
     * 3、创建输出对象
     */
    Text outKey = new Text();
    CovidCountBean outValue = new CovidCountBean();

    /**
     * 1、重写map父类方法：map回车
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        /**
         * 2、读取一行数据进行切割
         */
        String[] fields = value.toString().split(",");
        //["2020/1/21","Snohomish","Washington","1","0"]
        /**
         * 2020/1/21(时间),Snohomish(县),Washington(州),1(确诊数),0(死亡数)
         * 4、提取数据 州、确诊数、死亡数
         */
        String state = fields[2];
        outKey.set(state);
        //获得确诊数
        long covid_case = Long.parseLong(fields[3]);//"1" --> Long 1
        //获得死亡数
        long covid_death = Long.parseLong(fields[4]);
        outValue.set(covid_case,covid_death);

        /**
         * 5、输出结果
         */
        context.write(outKey,outValue);


    }
}
